package com.atguigu.chapter07;

import com.atguigu.chapter02.Event;
import com.atguigu.chapter05.Clicksource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

public class ProcessingTimeTimerTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new Clicksource()).keyBy( a -> true).
                process(new KeyedProcessFunction<Boolean, Event, String>() {
                    @Override
                    public void processElement(Event event, Context context, Collector<String> collector) throws Exception {
                        collector.collect(event.url+"到达 时间"+new Timestamp(context.timerService().currentProcessingTime()));
                        context.timerService().registerProcessingTimeTimer(10000L);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect( "定时器触发"+new Timestamp(ctx.timerService().currentProcessingTime()));
                    }
                }).print();



        env.execute();


    }
}
